package net.bwie.realtime.jtp.dwd.log.jjb;


import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

/**
 * @BelongsProject: realtime-project-10zlq
 * @BelongsPackage: net.bwie.realtime.jtp.dwd.log.jjb
 * @Author: zhangleqing
 * @CreateTime: 2025-08-17  19:33
 * @Description: TODO 分流
 * @Version: 1.0
 */
public class Text_02 extends ProcessFunction<String, String> {

    private final OutputTag<String> errorTag;
    private final OutputTag<String> startTag;
    private final OutputTag<String> displayTag;
    private final OutputTag<String> actionTag;

    public Text_02(OutputTag<String> errorTag,
                   OutputTag<String> startTag,
                   OutputTag<String> displayTag,
                   OutputTag<String> actionTag){
        this.errorTag = errorTag;
        this.startTag = startTag;
        this.displayTag = displayTag;
        this.actionTag = actionTag;
    }

    // value是输入的值  ctx是侧流  out是主流 我们写的reduce都是返回主流，侧流不做返回

    @Override
    public void processElement(String value,
                               Context ctx,
                               Collector<String> out) throws Exception {
        // 先解析
        JSONObject jsonObject = JSON.parseObject(value);
        // 全程不要; 不要短
        if (null != jsonObject.get("err")){
            // 放入侧流
            ctx.output(errorTag, value);
            jsonObject.remove("err");
        }
        if (null != jsonObject.get("start")){
            // todo 需要特别注意 容易出错
            ctx.output(startTag, jsonObject.toJSONString());
        }
        else {
            // 先获取 曝光和动作
            JSONArray displayJSONArray = jsonObject.getJSONArray("display");
            JSONArray actionJSONArray = jsonObject.getJSONArray("action");

            // 获取之后就进行删除
            jsonObject.remove("display");
            jsonObject.remove("action");

            out.collect(jsonObject.toJSONString());

            if (null != jsonObject.getJSONObject("display")& !jsonObject.getJSONObject("display").isEmpty()){
                for (int i = 0; i < displayJSONArray.size(); i++) {
                    JSONObject displayJSONObject = displayJSONArray.getJSONObject(i);
                    jsonObject.put("display", displayJSONObject);
                    ctx.output(displayTag, jsonObject.toJSONString());
                }
            }
            jsonObject.remove("display");
            if (null != jsonObject.getJSONObject("action")& !jsonObject.getJSONObject("action").isEmpty()){
                for (int i = 0; i < actionJSONArray.size(); i++) {
                    JSONObject actionJSONObject = actionJSONArray.getJSONObject(i);
                    jsonObject.put("action", actionJSONObject);
                    ctx.output(actionTag, jsonObject.toJSONString());
                }
            }
        }

    }


}
